package com.gitee.wsl.func.filter.bloomfilter.api

import com.gitee.wsl.base.Objects
import com.gitee.wsl.collections.list.toArray
import com.gitee.wsl.ext.coroutines.get
import com.gitee.wsl.ext.coroutines.set
import com.gitee.wsl.func.filter.bloomfilter.SimpleBloomFilter
import com.gitee.wsl.func.filter.bloomfilter.predicate.CountingPredicate
import com.gitee.wsl.func.predicate.BiPredicate
import com.gitee.wsl.func.predicate.Predicate
import kotlinx.atomicfu.atomic


/**
 * Produces Bloom filters from a collection (for example, [LayeredBloomFilter]).
 *
 * @since 4.5.0-M2
 */
fun interface BloomFilterExtractor<T : BloomFilter<T>> {
    /**
     * Return an array of the Bloom filters in the collection.
     *
     *
     * *Implementations should specify if the array contains deep copies, immutable instances, or references to the filters in the collection.*
     *
     *
     *
     * The default method returns a deep copy of the enclosed filters.
     *
     *
     * @return An array of Bloom filters.
     */
    fun asBloomFilterArray(): Array<out T> {
        val filters = ArrayList<T>()
        processBloomFilters(Predicate { f -> filters.add(f.copy()) })
        return filters.toArray() as Array<out T>
    }

    /**
     * Create a standard (non-layered) Bloom filter by merging all of the layers. If the filter is empty this method will return an empty Bloom filter.
     *
     * @return the merged bloom filter, never null.
     * @throws NullPointerException if this call did not process any filters.
     */
    fun flatten(): BloomFilter<T> {
        val ref = atomic<BloomFilter<T>?>(null)
        processBloomFilters(Predicate { x: T ->
            if (ref.get() == null) {
                ref.set(SimpleBloomFilter(x.shape) as BloomFilter<T>?)
            }
            ref.get()!!.merge(x)
        })
        return Objects.requireNonNull(ref.get(), "No filters.")
    }

    /**
     * Applies the `func` to each Bloom filter pair in order. Will apply all of the Bloom filters from the other BloomFilterExtractor to this extractor.
     * If either `this` extractor or `other` extractor has fewer BloomFilters the method will provide `null` for all excess calls to the
     * `func`.
     *
     *
     *
     * *This implementation returns references to the Bloom filter. Other implementations should specify if the array contains deep copies, immutable
     * instances, or references to the filters in the collection.*
     *
     *
     * @param other The other BloomFilterExtractor that provides the y values in the (x,y) pair.
     * @param func  The function to apply.
     * @return `true` if the `func` returned `true` for every pair, `false` otherwise.
     */
    fun processBloomFilterPair(other: BloomFilterExtractor<T>, func: BiPredicate<T?, T?>): Boolean {
        val p = CountingPredicate(asBloomFilterArray(), func)
        return other.processBloomFilters(p) && p.processRemaining()
    }

    /**
     * Executes a Bloom filter Predicate on each Bloom filter in the collection. The ordering of the Bloom filters is not specified by this interface.
     *
     * @param bloomFilterPredicate the predicate to evaluate each Bloom filter with.
     * @return `false` when the first filter fails the predicate test. Returns `true` if all filters pass the test.
     */
    fun processBloomFilters(bloomFilterPredicate: Predicate<T>): Boolean

    companion object {
        /**
         * Creates a BloomFilterExtractor from an array of Bloom filters.
         *
         *
         *  * The asBloomFilterArray() method returns a copy of the original array with references to the original filters.
         *  * The forEachBloomFilterPair() method uses references to the original filters.
         *
         *
         *
         * *All modifications to the Bloom filters are reflected in the original filters*
         *
         *
         * @param <T>     The BloomFilter type.
         * @param filters The filters to be returned by the extractor.
         * @return THe BloomFilterExtractor containing the filters.
        </T> */
        fun <T : BloomFilter<T>> fromBloomFilterArray(
            vararg filters: T
        ): BloomFilterExtractor<T> {


            return object : BloomFilterExtractor<T> {
                /**
                 * This implementation returns a copy the original array, the contained Bloom filters are references to the originals, any modifications to them are
                 * reflected in the original filters.
                 */
                override fun asBloomFilterArray(): Array<out T> {
                    return filters.clone()
                }

                /**
                 * This implementation uses references to the original filters. Any modifications to the filters are reflected in the originals.
                 */
                override fun processBloomFilterPair(
                    other: BloomFilterExtractor<T>,
                    func: BiPredicate<T?,T?>,
                ): Boolean {
                    val p = CountingPredicate(filters, func)
                    return other.processBloomFilters(p) && p.processRemaining()
                }

                override fun processBloomFilters(bloomFilterPredicate: Predicate<T>): Boolean {
                    for (filter in filters) {
                        if (!bloomFilterPredicate.test(filter)) {
                            return false
                        }
                    }
                    return true
                }
            }
        }
    }
}
